Coze 扩展包 PHP 版本

背景

最近在做项目的时候，想用 Coze 的 API，结果翻遍了官方文档，发现只有 Python、Node.js、Go 的 SDK，就是没有 PHP 的！

这让我有点郁闷，作为一个 PHP 开发者，总不能为了用个 API 就换语言吧？于是我就想，既然官方不给，那我自己搞一个呗！

安装

安装很简单，一行命令搞定：

bash

composer require pfinalclub/coze_sdk

基本要求也不高：

json

php: ^8.2
ext-http: *
ext-json: *
ext-openssl: *
firebase/php-jwt: ^6.10
psr/simple-cache: ^3.0
symfony/cache: ^7.1
symfony/http-client: ^7.1

目前版本是 v1.0.0，已经实现了这些功能：

功能展示

1. 聊天功能

这个是最基础的，用起来特别简单：

php

use CozeSdk\Chat\Chat;
// 创建聊天实例
$chat = new Chat($app);
// 发送消息
$response = $chat->setUserId('user_123')->sendMessage('你好，请介绍一下PHP');
// 处理响应
foreach ($response['messages'] as $message) {
    echo $message['content'] . "\n";
}

就像跟朋友聊天一样，设置个用户ID，然后直接发消息就行！

2. 机器人管理

有时候需要管理多个机器人，这个功能就派上用场了：

php

use CozeSdk\Bot\Bot;
// 创建机器人实例
$bot = new Bot($app);
// 获取机器人列表
$bots = $bot->getBots();
// 打印机器人列表
foreach ($bots as $bot) {
    echo $bot['name'] . "\n";
}

3. 基本流式传输

这个功能特别有意思，就像看直播一样，数据是实时流过来的：

php

use CozeSdk\Chat\Chat;

$chat = new Chat($app);

// 创建流式聊天（兼容旧版本）
$streamCallback = $chat->setUserId('user_123')
    ->Query('请写一个PHP函数')
    ->Build(true);

// 执行流式响应
$streamCallback();

4. 高级流式传输（推荐）

这个是我最得意的功能，提供了好几种处理方式，想怎么用就怎么用：

php

<?php

use CozeSdk\Chat\Chat;
use CozeSdk\Kernel\Support\{
    ConsoleStreamHandler,
    JsonStreamHandler,
    MemoryStreamHandler,
    CallbackStreamHandler
};

$chat = new Chat($app);

// 1. 控制台输出流式处理器 - 最简单，直接打印到控制台
$consoleHandler = new ConsoleStreamHandler();
$result = $chat->sendStreamMessage('你好，请介绍一下PHP', $consoleHandler);

// 2. JSON流式处理器 - 适合API接口，返回JSON格式
$jsonHandler = new JsonStreamHandler();
$result = $chat->sendStreamMessage('写一个简单的函数', $jsonHandler);

// 3. 内存流式处理器 - 把数据存在内存里，适合后续处理
$memoryHandler = new MemoryStreamHandler();
$result = $chat->sendStreamMessage('解释什么是OOP', $memoryHandler);

// 4. 回调流式处理器（最灵活）- 想怎么处理就怎么处理
$callbackHandler = new CallbackStreamHandler(
    function($chunk, $metadata) {
        // 处理每个数据块
        echo $chunk;
        flush();
        return true; // 继续处理
    },
    function($metadata) {
        // 流式传输开始
        echo "开始接收数据...\n";
    },
    function($metadata) {
        // 流式传输结束
        echo "\n接收完成\n";
    },
    function($error, $metadata) {
        // 错误处理
        echo "错误: " . $error->getMessage() . "\n";
    }
);

$result = $chat->sendStreamMessage('请写一个详细的教程', $callbackHandler);

5. 对话历史管理

在实际应用中，我们经常需要管理对话历史，这样AI才能理解上下文：

php

use CozeSdk\Chat\Chat;
use CozeSdk\Chat\Message;

$chat = new Chat($app);

// 创建对话历史
$history = [
    new Message('user', '我叫小明'),
    new Message('assistant', '你好小明，很高兴认识你！'),
    new Message('user', '我的名字是什么？')
];

// 发送带历史的消息
$response = $chat->setUserId('user_123')
    ->setHistory($history)
    ->sendMessage('请提醒我明天下午3点开会');

echo $response['messages'][0]['content'];

6. 自定义插件集成

Coze SDK 支持使用自定义插件来扩展功能：

php

use CozeSdk\Workflow\Workflow;

$workflow = new Workflow($app);

// 执行工作流
$result = $workflow->run([
    'workflow_id' => 'your_workflow_id',
    'parameters' => [
        'input' => '需要处理的数据',
        'options' => ['format' => 'json']
    ]
]);

// 获取工作流执行结果
if ($result['status'] === 'completed') {
    echo "执行成功：" . $result['output'];
} else {
    echo "执行失败：" . $result['error'];
}

高级用法

错误处理与重试机制

在生产环境中，网络不稳定是常见问题。SDK 内置了重试机制：

php

use CozeSdk\Chat\Chat;
use CozeSdk\Kernel\Exceptions\CozeSdkException;

$chat = new Chat($app);

try {
    // 设置重试次数和超时时间
    $response = $chat->setUserId('user_123')
        ->setRetry(3)  // 最多重试3次
        ->setTimeout(30)  // 超时30秒
        ->sendMessage('你好');
        
} catch (CozeSdkException $e) {
    // 处理SDK异常
    error_log("Coze SDK Error: " . $e->getMessage());
    
    // 根据错误类型进行不同处理
    switch ($e->getCode()) {
        case 401:
            echo "认证失败，请检查API密钥";
            break;
        case 429:
            echo "请求过于频繁，请稍后再试";
            break;
        case 500:
            echo "服务器错误，请联系管理员";
            break;
        default:
            echo "未知错误: " . $e->getMessage();
    }
}

缓存优化

为了提高性能，SDK 支持缓存机器人配置和常用响应：

php

use CozeSdk\Kernel\CacheManager;
use Symfony\Component\Cache\Adapter\RedisAdapter;

// 使用Redis作为缓存存储
$redis = new \Redis();
$redis->connect('127.0.0.1', 6379);

$cache = new RedisAdapter($redis);
$cacheManager = new CacheManager($cache);

// 启用缓存
$app->setCacheManager($cacheManager);

// 获取机器人信息（会自动缓存）
$bot = new Bot($app);
$botInfo = $bot->getBot('bot_id'); // 首次从API获取
$botInfo = $bot->getBot('bot_id'); // 第二次从缓存获取

批量操作

处理大量数据时，批量操作可以显著提高效率：

php

use CozeSdk\Chat\Chat;
use CozeSdk\Kernel\Support\BatchProcessor;

$chat = new Chat($app);
$processor = new BatchProcessor($chat);

// 批量发送消息
$messages = [
    ['user_id' => 'user_1', 'content' => '你好'],
    ['user_id' => 'user_2', 'content' => '早上好'],
    ['user_id' => 'user_3', 'content' => '晚上好']
];

// 并发处理（最多同时处理5个）
$results = $processor->batchSendMessages($messages, 5);

foreach ($results as $index => $result) {
    if ($result['success']) {
        echo "消息 {$index} 发送成功\n";
    } else {
        echo "消息 {$index} 发送失败: {$result['error']}\n";
    }
}

实战案例

案例1：智能客服系统

下面是一个完整的智能客服系统实现：

php

<?php

use CozeSdk\Chat\Chat;
use CozeSdk\Kernel\Support\CallbackStreamHandler;

class CustomerServiceBot
{
    private Chat $chat;
    private string $botId;
    
    public function __construct(Chat $chat, string $botId)
    {
        $this->chat = $chat;
        $this->botId = $botId;
    }
    
    public function handleCustomerQuery(string $userId, string $query): array
    {
        $responses = [];
        
        // 创建流式处理器
        $handler = new CallbackStreamHandler(
            onChunk: function($chunk, $metadata) use (&$responses) {
                $responses[] = $chunk;
                return true;
            },
            onError: function($error) {
                error_log("客服机器人错误: " . $error->getMessage());
            }
        );
        
        // 发送流式消息
        $this->chat->setUserId($userId)
            ->setBotId($this->botId)
            ->sendStreamMessage($query, $handler);
        
        return [
            'user_id' => $userId,
            'query' => $query,
            'response' => implode('', $responses),
            'timestamp' => time()
        ];
    }
    
    public function getCustomerHistory(string $userId): array
    {
        // 从数据库获取历史记录
        return $this->chat->getHistory($userId);
    }
}

// 使用示例
$bot = new CustomerServiceBot($chat, 'your_bot_id');
$result = $bot->handleCustomerQuery('user_123', '我想退货');

echo "客服回复: " . $result['response'];

案例2：内容生成器

使用 Coze SDK 批量生成营销文案：

php

<?php

use CozeSdk\Chat\Chat;

class ContentGenerator
{
    private Chat $chat;
    
    public function generateProductDescriptions(array $products): array
    {
        $descriptions = [];
        
        foreach ($products as $product) {
            $prompt = "请为以下产品生成吸引人的描述：\n";
            $prompt .= "产品名称：{$product['name']}\n";
            $prompt .= "产品特点：" . implode(', ', $product['features']) . "\n";
            $prompt .= "目标受众：{$product['target_audience']}\n";
            
            try {
                $response = $this->chat->setUserId('system')
                    ->setTimeout(60)
                    ->sendMessage($prompt);
                
                $descriptions[$product['id']] = [
                    'title' => $product['name'],
                    'description' => $response['messages'][0]['content'],
                    'status' => 'success'
                ];
            } catch (\Exception $e) {
                $descriptions[$product['id']] = [
                    'title' => $product['name'],
                    'description' => '',
                    'status' => 'failed',
                    'error' => $e->getMessage()
                ];
            }
            
            // 避免请求过于频繁
            usleep(500000); // 休眠0.5秒
        }
        
        return $descriptions;
    }
}

性能优化建议

1. 连接池管理

对于高并发场景，建议使用连接池：

php

use CozeSdk\Kernel\ConnectionPool;

$pool = new ConnectionPool([
    'min_connections' => 5,
    'max_connections' => 20,
    'wait_timeout' => 3.0
]);

$app->setConnectionPool($pool);

2. 请求限流

防止API配额耗尽：

php

use CozeSdk\Kernel\RateLimiter;

$limiter = new RateLimiter([
    'max_requests' => 100,  // 每分钟最多100个请求
    'time_window' => 60      // 时间窗口60秒
]);

if ($limiter->allowRequest()) {
    $response = $chat->sendMessage('你好');
} else {
    echo "请求过于频繁，请稍后再试";
}

3. 异步处理

对于非实时场景，使用队列异步处理：

php

use CozeSdk\Kernel\Queue\QueueManager;

$queue = new QueueManager('redis', [
    'host' => '127.0.0.1',
    'port' => 6379
]);

// 添加到队列
$queue->push('chat_queue', [
    'user_id' => 'user_123',
    'message' => '你好'
]);

// 处理队列
$queue->process('chat_queue', function($job) use ($chat) {
    $response = $chat->setUserId($job['user_id'])
        ->sendMessage($job['message']);
    return $response;
});

故障排查

常见问题

1. 认证失败

Error: Invalid API Key

解决方案：检查 .env 文件中的 API Key 是否正确

2. 超时错误

Error: Request timeout after 30 seconds

解决方案：增加超时时间或优化网络环境

3. 流式传输中断

Error: Stream interrupted

解决方案：检查网络稳定性，实现自动重连机制

调试技巧

启用调试模式查看详细日志：

php

use CozeSdk\Kernel\Logger;

$logger = new Logger('coze_debug.log', Logger::DEBUG);
$app->setLogger($logger);

// 所有请求和响应都会被记录
$response = $chat->sendMessage('测试消息');

最佳实践

环境变量管理：使用 .env 文件管理敏感信息
错误处理：始终使用 try-catch 捕获异常
日志记录：记录所有API调用和错误信息
缓存策略：合理使用缓存减少API调用
限流保护：实现请求限流避免超出配额
监控告警：监控API调用状态和错误率

使用心得

开发这个扩展包的过程让我对Coze API有了更深入的理解。在实际项目中使用后，发现流式传输功能特别适合做聊天机器人，用户体验比传统的等待完整响应要好很多。

在生产环境中，建议：

使用Redis缓存常用响应
实现请求队列处理大批量任务
配置完善的日志和监控系统
定期备份对话历史数据

后续计划

功能增强
- 支持更多Coze API功能
- 添加WebSocket支持实现双向通信
- 实现更智能的错误重试策略
性能优化
- 优化内存使用
- 提升并发处理能力
- 减少网络延迟
文档完善
- 添加更多使用示例
- 提供视频教程
- 建立社区讨论区
生态建设
- Laravel集成包
- Symfony Bundle
- WordPress插件

社区与支持

GitHub: pfinalclub/coze_sdk
文档: https://docs.pfinal.club/coze-sdk
问题反馈: GitHub Issues
技术交流: 微信群（扫码加入）

致谢

感谢所有为这个项目贡献代码和提出建议的开发者。特别感谢 Coze 团队提供如此优秀的API服务。

如果这个扩展包对你有帮助，欢迎 Star ⭐ 支持！

作者：PFinal南丞 - 一个热爱PHP和开源的程序员
最后更新：2025-11-10

背景 ​

安装 ​

功能展示 ​

1. 聊天功能 ​

2. 机器人管理 ​

3. 基本流式传输 ​

4. 高级流式传输（推荐） ​

5. 对话历史管理 ​

6. 自定义插件集成 ​

高级用法 ​

错误处理与重试机制 ​

缓存优化 ​

批量操作 ​

实战案例 ​

案例1：智能客服系统 ​

案例2：内容生成器 ​

性能优化建议 ​

1. 连接池管理 ​

2. 请求限流 ​

3. 异步处理 ​

故障排查 ​

常见问题 ​

调试技巧 ​

最佳实践 ​

使用心得 ​

后续计划 ​

社区与支持 ​

致谢 ​

背景

安装